package com.lj.hadoop;

import java.io.IOException;
import java.net.URI;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

/**
 * 测试成功
 * @author Administrator
 *
 */
public class MysqlToHadoop {

	/**
	 * map作业读取数据记录数
	 */

	private static class DBMap extends
			Mapper<LongWritable, DSDATAUSERRecord, LongWritable, Text> {

		@Override
		protected void map(LongWritable key, DSDATAUSERRecord value,
				Context context) throws IOException, InterruptedException {
			// 使用mapper写数据
			context.write(new LongWritable(value.getId()),
					new Text(value.toString()));

		}

	}

	/**
	 * 由于数据库中有主键的数据是唯一的，所以
	 * 
	 * @author Administrator
	 * 
	 */
	public static class DBReducer extends
			Reducer<LongWritable, Text, LongWritable, Text> {
		@Override
		public void reduce(LongWritable key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException {
			for (Iterator<Text> itr = values.iterator(); itr.hasNext();) {
				context.write(key, itr.next());
			}
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		MysqlToHadoop test = new MysqlToHadoop();
		// 我们通过spring的方式获取到hadoop的配置信息
		ApplicationContext ctx = new ClassPathXmlApplicationContext(
				"classpath:spring/applicationContext.xml");
		// 得到hadoop的配置
		Configuration config = ctx.getBean("hadoopConfiguration",
				Configuration.class);
		
		System.out.println(config.get("fs.defaultFS"));
		Path output = new Path("/test/user.txt");
		FileSystem fs = FileSystem.get(URI.create(output.toString()), config);
		if (fs.exists(output)) {
			System.out.println("删除文件路径：" + output.toString());
			fs.delete(output, true);
		}
		
		
		DBConfiguration
				.configureDB(
						config,
						"com.mysql.jdbc.Driver",
						"jdbc:mysql://172.16.135.160/DSDATA?useUnicode=true&characterEncoding=UTF-8",
						"root", "123456");
		Job job = Job.getInstance(config, "这是一个读取mysql数据的测试");
		// 通过类名找到jar包
		job.setJarByClass(MysqlToHadoop.class);
		// 这里没有必要设置一个combine类来优化
		job.setMapperClass(DBMap.class);
		job.setReducerClass(DBReducer.class);
		// 设置输出的key类型
		job.setOutputKeyClass(LongWritable.class);
		// 设置输出的value类型
		job.setOutputValueClass(Text.class);
		// 设置输入规格类型
		job.setInputFormatClass(DBInputFormat.class);
		// 设置任务的输出路径
		FileOutputFormat.setOutputPath(job, output);
		/*FileOutputFormat.*/
		// 下面为列名
		String[] fields = { "ID", "USER_NAME", "PASSWORD", "DISTRICT_NAME",
				"START_AUDIT", "END_AUDIT", "DESCRIPTION", "UPPER_DISTRICT" };

		// 六个参数分别为：
		// 1.Job;2.Class<? extends DBWritable>
		// 3.表名;4.where条件
		// 5.order by语句;6.列名
		DBInputFormat.setInput(job, DSDATAUSERRecord.class, "USER",
				null, "ID", fields);
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		
		//成功后生成的文件

	}

}
